//package com.caishi.lkx.user.service.impl;
//
//import cn.hutool.core.bean.BeanUtil;
//import cn.hutool.core.collection.CollUtil;
//import cn.hutool.core.date.LocalDateTimeUtil;
//import cn.hutool.core.map.MapUtil;
//import cn.hutool.core.util.RandomUtil;
//import cn.hutool.core.util.StrUtil;
//import com.alibaba.fastjson.JSON;
//import com.caishi.lkx.common.util.IpAuthUtil;
//import com.caishi.lkx.user.ienum.type.GroupLogRuleTimeType;
//import com.caishi.lkx.user.ienum.type.UserLogSourceType;
//import com.caishi.lkx.user.mapper.SysGroupRuleMapper;
//import com.zzw.common.Wrappers;
//import com.zzw.common.entity.IBaseMapper;
//import com.zzw.common.entity.impl.BaseIntEntity;
//import com.zzw.common.utils.TimeUtil;
//
//import com.caishi.lkx.user.model.GroupLogRuleModel;
//import com.caishi.lkx.user.service.IGroupLogRuleService;
//import com.caishi.lkx.user.userlog.FalsificationImageRepository;
//import com.caishi.lkx.user.userlog.NormalLogRecordsService;
//import com.caishi.lkx.user.userlog.model.FalsificationImage;
//import com.caishi.lkx.user.userlog.model.NormalLogRecords;
//import com.caishi.lkx.user.userlog.model.UserLogType;
//import lombok.Getter;
//import lombok.Setter;
//import lombok.extern.slf4j.Slf4j;
//import org.apache.rocketmq.client.producer.SendCallback;
//import org.apache.rocketmq.client.producer.SendResult;
//import org.apache.rocketmq.spring.annotation.MessageModel;
//import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
//import org.apache.rocketmq.spring.core.RocketMQListener;
//import org.apache.rocketmq.spring.core.RocketMQTemplate;
//import org.springframework.stereotype.Service;
//
//import javax.annotation.PreDestroy;
//import javax.annotation.Resource;
//import java.time.LocalDate;
//import java.time.LocalDateTime;
//import java.time.LocalTime;
//import java.time.format.DateTimeFormatter;
//import java.util.*;
//import java.util.concurrent.ScheduledExecutorService;
//import java.util.concurrent.ScheduledThreadPoolExecutor;
//import java.util.concurrent.TimeUnit;
//
//@Service("groupRuleService")
//@Slf4j
//@RocketMQMessageListener(topic = "vrs7-log-rule-stop", consumerGroup = "${rocketmq.producer.group}-log", messageModel = MessageModel.BROADCASTING)
//public class GroupLogRuleServiceImpl implements IGroupLogRuleService, RocketMQListener<GroupLogRuleServiceImpl.TaskMqDto> {
//
//    @Getter
//    @Setter
//    public static class TaskMqDto {
//        private boolean stop;
//        private Long groupId;
//        private Long id;
//    }
//
//
//    private static final int SCAN_SECOND = 3600;
//
//    @Resource
//    private SysGroupRuleMapper groupRuleMapper;
//
//    @Resource
//    private RocketMQTemplate rocketMQTemplate;
//
//    @Resource
//    private NormalLogRecordsService recordsService;
//
//    @Resource
//    private FalsificationImageRepository falsificationImageRepository;
//
//    private static final Map<Long, Set<GroupLogRuleModel>> TASK = new HashMap<>();
//
//    private final ScheduledExecutorService schedulingException = new ScheduledThreadPoolExecutor(100, r -> {
//        Thread t = new Thread(r);
//        t.setName("log-task-pool");
//        return t;
//    });
//
//    @Override
//    public IBaseMapper<GroupLogRuleModel, Long> getMapper() {
//        return groupRuleMapper;
//    }
//
//    @Override
//    public void onMessage(TaskMqDto dto) {
//        if (dto.stop) mqCloseTask(dto.getId(), dto.groupId);
//        else mqContinueTask(dto.getId());
//    }
//
////    @Scheduled(fixedDelay = SCAN_SECOND * 1000)
//    public void init() {
//        scanTask();
//    }
//
//    @PreDestroy
//    public void close() {
//        log.info("关闭日志记录缓存任务");
//        schedulingException.shutdownNow();
//    }
//
//    @Override
//    public GroupLogRuleModel insert(GroupLogRuleModel entity) {
//        entity = IGroupLogRuleService.super.insert(entity);
//        if (entity.getTimeType() == GroupLogRuleTimeType.future) {
//            continueRule(entity.getId());
//        } else {
//            entity.setStatus(2);
//            addChangeBeforeTask(entity);
//        }
//        return entity;
//    }
//
//    @Override
//    public GroupLogRuleModel update(GroupLogRuleModel entity) {
//        return simpleUpdate(entity);
//    }
//
//    @Override
//    public GroupLogRuleModel simpleUpdate(GroupLogRuleModel entity) {
//        groupRuleMapper.updateById(entity);
//        if (entity.getTimeType() == GroupLogRuleTimeType.future) {
//            if (!timeWork(entity.getTimeMode(), entity.getStart(), entity.getEnd())) {
//                stopRule(entity.getId(), 1);
//            }
//            continueRule(entity.getId());
//        }
//        return entity;
//    }
//
//    @Override
//    public double getMachiningCount(Long groupId, UserLogType logType, UserLogSourceType userLogSourceType, String industryId, String userId, String ip, double realData) {
//        var runningRule = TASK.get(groupId);
//        if (CollUtil.isEmpty(runningRule)) return realData;
//        var rule = runningRule.stream().filter(rl -> {
//            var r = rl.getSourceTypes().contains(userLogSourceType);
//            r = r && (CollUtil.isEmpty(rl.getUserIds()) || rl.getUserIds().contains(userId));
//            r = r && timeWork(rl.getTimeMode(), rl.getStart(), rl.getEnd());
//            return r && (CollUtil.isEmpty(rl.getIps()) || IpAuthUtil.ipInIps(ip, rl.getIps()));
//        }).findFirst().orElse(null);
//        if (rule == null) return realData;
//        var ruleData = rule.getRule();
//        if (MapUtil.isEmpty(ruleData)) return realData;
//        var typeRule = ruleData.get(logType);
//        if (MapUtil.isEmpty(typeRule)) return realData;
//        var industryRule = typeRule.get(industryId);
//        if (StrUtil.isBlank(industryRule)) {
//            industryRule = typeRule.get("all");
//        }
//        if (StrUtil.isBlank(industryRule)) return realData;
//        var ks = industryRule.split("_");
//        var type = Integer.parseInt(ks[0]);
//        var typeValue = Double.parseDouble(ks[1]);
//        if (type == 0) return realData + realData * typeValue;
//        return typeValue + realData;
//    }
//
//    @Override
//    public void stopRule(Long id, int status) {
//        var task = getById(id);
//        task.setStatus(status);
//        groupRuleMapper.updateById(task);
//        var dto = new TaskMqDto();
//        dto.setStop(true);
//        dto.setGroupId(task.getGroupId());
//        dto.setId(task.getId());
//        rocketMQTemplate.asyncSend("vrs7-log-rule-stop", dto, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                log.info("规则停止成功 {}", id);
//            }
//
//            @Override
//            public void onException(Throwable throwable) {
//                log.error("send message failed:", throwable);
//                throwable.printStackTrace();
//            }
//        });
//    }
//
//    @Override
//    public void continueRule(Long id) {
//        var dto = new TaskMqDto();
//        dto.setStop(false);
//        dto.setId(id);
//        rocketMQTemplate.asyncSend("vrs7-log-rule-stop", dto, new SendCallback() {
//            @Override
//            public void onSuccess(SendResult sendResult) {
//                log.info("规则继续成功 {}", id);
//            }
//
//            @Override
//            public void onException(Throwable throwable) {
//                log.error("send message failed:", throwable);
//                throwable.printStackTrace();
//            }
//        });
//    }
//
//    /**
//     * 添加修改过往数据任务
//     *
//     * @param model
//     */
//    public void addChangeBeforeTask(GroupLogRuleModel model) {
//        var rule = model.getRule();
//        rule.forEach((type, value) -> changeUserLogTypeValue(model.getId(), type, model.getGroupId(), model.getSourceTypes(), model.getTimeType(), LocalDate.parse(model.getStart(), TimeUtil.DATE_TIME_FORMATTER_DAY),
//                LocalDate.parse(model.getEnd(), TimeUtil.DATE_TIME_FORMATTER_DAY),
//                value
//        ));
//        model.setStatus(3);
//        getMapper().updateById(model);
//    }
//
//    public void mqCloseTask(Long id, Long groupId) {
//        log.info("mq关闭任务 {}", id);
//        var tasks = TASK.get(groupId);
//        if (tasks != null) {
//            var r = tasks.stream().filter(v -> v.getId().equals(id)).findFirst().orElse(null);
//            if (r != null) {
//                tasks.remove(r);
//                r.setDeleted(true);
//            }
//        }
//    }
//
//    public void mqContinueTask(Long id) {
//        log.info("mq开始任务 {}", id);
//        var task = getById(id);
//        if (task != null) {
//            task.setStatus(1);
//            addStarTask(task);
//        }
//    }
//
//
//    private void scanTask() {
//        var tasks = selectList(wrappers().eq(GroupLogRuleModel::getTimeType, GroupLogRuleTimeType.future)
//                .in(GroupLogRuleModel::getStatus, 1, 2, 5)
//        );
//        for (var task : tasks) {
//            // 如果任务本来就是执行中 直接开始任务
//            if (task.getStatus() == 2) {
//                startTask(task);
//            } else {
//                addStarTask(task);
//            }
//        }
//    }
//
//    private void startTask(GroupLogRuleModel model) {
//        if (model.getDeleted()) return;
//        log.info("开始运行注水规则 id={} status={} s={} e={} t={}", model.getId(), model.getStatus(), model.getStart(), model.getEnd(), model.getTimeMode());
//        // 如果任务不在执行时间里 关闭任务
//        if (!timeWork(model.getTimeMode(), model.getStart(), model.getEnd())) {
//            stopTask(model);
//            return;
//        }
//        // 修改任务状态为执行中
//        if (model.getStatus() != 2) {
//            model.setStatus(2);
//            groupRuleMapper.update(null, Wrappers.<GroupLogRuleModel>lambdaUpdate()
//                    .set(GroupLogRuleModel::getStatus, 2)
//                    .eq(BaseIntEntity::getId, model.getId())
//            );
//        }
//        var tasks = TASK.get(model.getGroupId());
//        // 任务以前不在运行中 添加任务
//        if (tasks == null) {
//            TASK.put(model.getGroupId(), new LinkedHashSet<>() {{
//                add(model);
//            }});
//        } else {
//            var old = tasks.stream().filter(v -> Objects.equals(v.getId(), model.getId())).findFirst().orElse(null);
//            if (old == null) {
//                tasks.add(model);
//            } else {
//                BeanUtil.copyProperties(model, old, "id");
//            }
//            closeTask(model);
//        }
//    }
//
//    private void stopTask(GroupLogRuleModel model) {
//        if (model.getDeleted()) return;
//        log.info("停止运行注水规则 id={} status={} s={} e={} t={}", model.getId(), model.getStatus(), model.getStart(), model.getEnd(), model.getTimeMode());
//        model.setStatus(1);
//        groupRuleMapper.update(null, Wrappers.<GroupLogRuleModel>lambdaUpdate()
//                .set(GroupLogRuleModel::getStatus, 1)
//                .eq(BaseIntEntity::getId, model.getId())
//        );
//        var tasks = TASK.get(model.getGroupId());
//        // 移除任务
//        if (CollUtil.isNotEmpty(tasks)) tasks.remove(model);
//        // 任务移除后判断下次开始
//        addStarTask(model);
//    }
//
//    /**
//     * 修改机构指定日志类型在时间段的值
//     *
//     * @param logType
//     * @param groupId
//     * @param timeType
//     * @param start
//     * @param end
//     * @param indType
//     */
//    private void changeUserLogTypeValue(long ruleId, UserLogType logType, Long groupId, List<UserLogSourceType> sourceTypes, GroupLogRuleTimeType timeType, LocalDate start, LocalDate end, Map<String, String> indType) {
//        if (logType.isInd()) {
//            changeUserLogTypeValueAtIndustry(ruleId, logType, groupId, sourceTypes, timeType, start, end, indType);
//            return;
//        }
//        var image = new FalsificationImage();
//        image.setGroupId(groupId);
//        image.setRuleId(ruleId);
//        var records = recordsService.groupRecordList(groupId, logType, start, end);
//        var valueRule = indType.get("all");
//        var ks = valueRule.split("_");
//        var allChange = changeValue(image, records, logType, sourceTypes, timeType, Integer.parseInt(ks[0]), Double.parseDouble(ks[1]));
//        if (end.isBefore(LocalDate.now())) {
//            allChange.forEach((t, v) -> recordsService.updateRecordValueAtBeforeTime(logType, groupId, null, end, t, v));
//        }
//        falsificationImageRepository.insert(image);
//    }
//
//    private void changeUserLogTypeValueAtIndustry(long ruleId, UserLogType logType, Long groupId, List<UserLogSourceType> sourceTypes, GroupLogRuleTimeType timeType, LocalDate start, LocalDate end, Map<String, String> indType) {
//        // 如果所有 不区分行业。将对所有行业都加上这个数字
//        var image = new FalsificationImage();
//        image.setGroupId(groupId);
//        image.setRuleId(ruleId);
//        if (indType.containsKey("all")) {
//            var ids = recordsService.allIndustry(groupId);
//            var valueRule = indType.get("all");
//            var ks = valueRule.split("_");
//            ids.forEach(id -> {
//                var allChange = changeValue(image, recordsService.groupIndustryRecordList(groupId, logType, id, start, end), logType, sourceTypes, timeType, Integer.parseInt(ks[0]), Double.parseDouble(ks[1]));
//                allChange.forEach((t, v) -> recordsService.updateRecordValueAtBeforeTime(logType, groupId, id, end, t, v));
//            });
//        } else {
//            indType.forEach((id, valueRule) -> {
//                var ks = valueRule.split("_");
//                var allChange = changeValue(image, recordsService.groupIndustryRecordList(groupId, logType, id, start, end), logType, sourceTypes, timeType, Integer.parseInt(ks[0]), Double.parseDouble(ks[1]));
//                allChange.forEach((t, v) -> recordsService.updateRecordValueAtBeforeTime(logType, groupId, id, end, t, v));
//            });
//        }
//        falsificationImageRepository.insert(image);
//    }
//
//    private Map<UserLogSourceType, Double> changeValue(FalsificationImage image, List<? extends NormalLogRecords> records, UserLogType logType, List<UserLogSourceType> sourceTypes, GroupLogRuleTimeType timeType, int type, double typeValue) {
//        // 随机分配
//        image.getBeforeJson().add(JSON.toJSONString(records));
//        records.sort(Comparator.comparing(NormalLogRecords::getDateT));
//        Map<UserLogSourceType, Double> result = new HashMap<>();
//        sourceTypes.forEach(v -> result.put(v, 0.0));
//        if (timeType == GroupLogRuleTimeType.pastRandom) {
//            // 随机给每天分配数量
//            var values = generateValue(records.size(), typeValue);
//            for (var i = 0; i < records.size(); i++) {
//                var count = records.get(i).getData();
//                // 后面的把前面注水的数据加上
//                result.forEach((t, v) -> count.addAllCount(t, v, false));
//                // 随机给平台分配数量
//                var validSourceTypes = new ArrayList<UserLogSourceType>() {{
//                    addAll(CollUtil.intersection(sourceTypes, count.validSource()));
//                }};
//                var tv = generateValue(validSourceTypes.size(), values[i]);
//                for (var j = 0; j < validSourceTypes.size(); j++) {
//                    var t = validSourceTypes.get(j);
//                    result.put(t, result.get(t) + tv[j]);
//                    count.addAllCount(t, tv[j], true);
//                }
//                recordsService.updateRecord(logType, records.get(i));
//            }
//        } else {
//            // 平均分配
//            if (type == 1) {
//                // 固定值分配是平均加值
//                var itemValue = typeValue / records.size();
//                for (var rd : records) {
//                    var count = rd.getData();
//                    // 后面的把前面注水的数据加上
//                    result.forEach((t, v) -> count.addAllCount(t, v, false));
//                    var validSourceTypes = CollUtil.intersection(sourceTypes, count.validSource());
//                    var tv = itemValue / validSourceTypes.size();
//                    for (var st : validSourceTypes) {
//                        result.put(st, result.get(st) + tv);
//                        count.addAllCount(st, tv, true);
//                    }
//                    recordsService.updateRecord(logType, rd);
//                }
//            } else {
//                // 固定倍数时将每个有效总数加上倍数后的值
//                for (var rd : records) {
//                    var count = rd.getData();
//                    // 后面的把前面注水的数据加上
//                    result.forEach((t, v) -> count.addAllCount(t, v, false));
//                    var validSourceTypes = CollUtil.intersection(sourceTypes, count.validSource());
//                    for (var st : validSourceTypes) {
//                        var k = count.getTypeItem(st).getAllCount() * typeValue;
//                        result.put(st, result.get(st) + k);
//                        count.addAllCount(st, k, true);
//                    }
//                    recordsService.updateRecord(logType, rd);
//                }
//            }
//        }
//        // 保存注水后的数据
//        image.getAfterJson().add(JSON.toJSONString(records));
//        return result;
//    }
//
//    private double[] generateValue(int size, double sum) {
//        if (size == 1) return new double[]{sum};
//        if (size == 2) {
//            var k = RandomUtil.randomDouble(0, sum);
//            return new double[]{k, sum - k};
//        }
//        double[] result = new double[size];
//        int[] random = new int[size];
//        double randomSum = 0;
//        for (var i = 0; i < size; i++) {
//            var k = RandomUtil.randomInt(1, 100);
//            random[i] = k;
//            randomSum += k;
//        }
//        assert randomSum > 0;
//        for (var i = 0; i < size; i++) {
//            result[i] = sum * (random[i] / randomSum);
//        }
//        return result;
//    }
//
//
//    /**
//     * 添加将要开始的任务
//     *
//     * @param model
//     */
//    private void addStarTask(GroupLogRuleModel model) {
//        if (model.getDeleted() || model.getStatus() != 1) return;
//        var now = LocalDateTime.now();
//        var start = getTime(now, model.getTimeMode(), model.getStart());
//        if (timeWork(model.getTimeMode(), model.getStart(), model.getEnd())) {
//            startTask(model);
//            return;
//        }
//        if (start.minusSeconds(SCAN_SECOND).isBefore(now) && start.isAfter(now)) {
//            log.info("提交开始运行注水规则 id={} status={} s={} e={} t={}", model.getId(), model.getStatus(), model.getStart(), model.getEnd(), model.getTimeMode());
//            schedulingException.schedule(() -> startTask(model), LocalDateTimeUtil.toEpochMilli(start) - LocalDateTimeUtil.toEpochMilli(now), TimeUnit.MILLISECONDS);
//        }
//    }
//
//    /**
//     * 关闭将要结束的任务
//     *
//     * @param model
//     */
//    private void closeTask(GroupLogRuleModel model) {
//        if (model.getDeleted() || timeWork(model.getTimeMode(), model.getStart(), model.getEnd())) return;
//        if (model.getStatus() != 2) return;
//        var now = LocalDateTime.now();
//        var end = getTime(now, model.getTimeMode(), model.getEnd());
//        if (end.isBefore(now) || end.equals(now)) {
//            stopTask(model);
//            return;
//        }
//        if (end.minusSeconds(SCAN_SECOND).isBefore(now) && end.isAfter(now)) {
//            log.info("提交关闭运行注水规则 id={} status={} s={} e={} t={}", model.getId(), model.getStatus(), model.getStart(), model.getEnd(), model.getTimeMode());
//            schedulingException.schedule(() -> stopTask(model), LocalDateTimeUtil.toEpochMilli(end) - LocalDateTimeUtil.toEpochMilli(now), TimeUnit.MILLISECONDS);
//        }
//    }
//
//
//    /**
//     * 判断是否在执行期间
//     *
//     * @param mode
//     * @param start
//     * @param end
//     * @return
//     */
//    private boolean timeWork(int mode, String start, String end) {
//        // 0 一段 1每年 2每月 3每日 4 每小时 5每分
//        var now = LocalDateTime.now();
//        LocalDateTime s = getTime(now, mode, start);
//        LocalDateTime e = getTime(now, mode, end);
//        if (s == null) return false;
//        return now.isBefore(e) && (now.isAfter(s) || now.equals(s));
//    }
//
//    private LocalDateTime getTime(LocalDateTime now, int mode, String time) {
//        LocalDateTime s = null;
//        if (mode == 0) {
//            s = LocalDateTime.parse(time, TimeUtil.DATE_TIME_FORMATTER_SC);
//        } else if (mode == 1) {
//            s = LocalDateTime.parse(time, TimeUtil.DATE_TIME_FORMATTER_SC);
//            s = s.plusYears(now.getYear() - s.getYear());
//        } else if (mode == 2) {
//            s = LocalDateTime.parse(time, TimeUtil.DATE_TIME_FORMATTER_SC);
//            s = s.plusYears(now.getYear() - s.getYear()).plusMonths(now.getMonthValue() - s.getMonthValue());
//        } else if (mode == 3) {
//            var s1 = LocalTime.parse(time, DateTimeFormatter.ofPattern("HH:mm:ss"));
//            s = s1.atDate(now.toLocalDate());
//        } else if (mode == 4) {
//            var s1 = LocalTime.parse(time, DateTimeFormatter.ofPattern("HH:mm:ss"));
//            s = LocalTime.of(now.getHour(), s1.getMinute(), s1.getSecond()).atDate(now.toLocalDate());
//        } else if (mode == 5) {
//            var s1 = LocalTime.parse(time, DateTimeFormatter.ofPattern("HH:mm:ss"));
//            s = LocalTime.of(now.getHour(), now.getMinute(), s1.getSecond()).atDate(now.toLocalDate());
//        }
//        return s;
//    }
//
//}
